package me.tuyou.function;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple3;

public class AggregateTest {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Tuple3<Integer, String, Double>> input = env.fromElements(
                Tuple3.of(1, "a", 1.0),
                Tuple3.of(1, "a", 1.1),
                Tuple3.of(2, "b", 2.2),
                Tuple3.of(2, "b", 2.3),
                Tuple3.of(3, "c", 3.3),
                Tuple3.of(3, "c", 3.4)
        );

        AggregateOperator<Tuple3<Integer, String, Double>> output1 = input.groupBy(1)
                .aggregate(Aggregations.SUM, 0)
                .and(Aggregations.MIN, 2);

        AggregateOperator<Tuple3<Integer, String, Double>> output2 = input.groupBy(1)
                .aggregate(Aggregations.SUM, 0)
                .aggregate(Aggregations.MIN, 2);

        System.out.println("output1----");
        output1.print();

        System.out.println("output2----");
        output2.print();
    }
}
